package com.youxin.foundation.zookeeper;


import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.Ids;


import org.apache.zookeeper.ZooKeeper;

import com.youxin.foundation.utils.thread.ScheduleThreadPool;
import com.youxin.foundation.utils.timer.TimerSchedule;

/**
 * Zookeeper连接器
 * @author yingde.cao
 *
 */
public class ZookeeperConnector {
	public static final int ZOOKEEPER_TIMEOUT=8000;
    private ZooKeeper zk;
    private boolean isConnected=false;

    public ZooKeeper connect(String host) throws Exception {    	
        CountDownLatch connSignal = new CountDownLatch(1);
        ScheduleThreadPool.schedule(new Runnable() {			
			@Override
			public void run() {				
	            connSignal.countDown();				
			}
		}, ZOOKEEPER_TIMEOUT);        

        zk = new ZooKeeper(host, ZOOKEEPER_TIMEOUT, new Watcher() {
        	@Override
            public void process(WatchedEvent event) {
                if (event.getState() == KeeperState.SyncConnected) {
                	isConnected=true;
                }
                connSignal.countDown();
            }
        });        
        connSignal.await();
        return zk;
    }
    
    

    public boolean isConnected() {
		return isConnected;
	}



	public void setConnected(boolean isConnected) {
		this.isConnected = isConnected;
	}



	public void close() throws InterruptedException {
        zk.close();
    }

    public void createNode(String path, byte[] data) throws Exception
    {
        zk.create(path, data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }

    public void updateNode(String path, byte[] data) throws Exception
    {
        zk.setData(path, data, zk.exists(path, true).getVersion());
    }

    public void deleteNode(String path) throws Exception
    {
        zk.delete(path,  zk.exists(path, true).getVersion());
    }    
}